package jk.fox.node;

import jk.fox.common.RpcDict;
import jk.fox.common.exception.runtime.RpcRunException;
import jk.fox.common.payload.message.Message;
import jk.fox.common.payload.message.MessageDispatcher;
import jk.fox.common.payload.response.RpcResponse;
import jk.fox.common.payload.server.DefaultServer;
import jk.fox.exchange.server.ServerExchange;
import jk.fox.exchange.client.ClientExchange;
import jk.fox.transport.ClientTransporter;
import jk.fox.transport.ServerTransporter;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;

import java.net.InetSocketAddress;


/**
 * @author cuichao
 * @Description: DefaultServerMessageDispatcher
 * @Date: create in 2020/5/15 20:31
 */
@Setter
@Getter
@NoArgsConstructor
@AllArgsConstructor
public class StandardMessageDispatcher implements MessageDispatcher {

    private volatile ServerExchange serverExchange;

    private ServerTransporter serverTransporter;

    private ClientExchange clientExchange;

    private ClientTransporter clientTransporter;

    @Override
    public Message dispatch(Message message) {
        switch (message.getStatus()){
            case Init:
                initServerPostProcess(message);
                break;
            case Invalid:
                reconnectPostProcess(message);
                break;
            case Request:
                receivedRequestPostProcess(message);
                break;
            case Response:
                break;
            default:
                break;
        }
        return message;
    }

    @Override
    public void clientExchange(ClientExchange exchange) {
        this.clientExchange = exchange;
    }

    @Override
    public void clientTransporter(ClientTransporter transporter) {
        this.clientTransporter = transporter;
    }

    @Override
    public void serverExchange(ServerExchange exchange) {
        this.serverExchange = exchange;
    }

    @Override
    public void serverTransporter(ServerTransporter transporter) {
        this.serverTransporter = transporter;
    }

    /**
     * 服务端转发消息
     * @param message
     * @return
     */
    private void receivedRequestPostProcess(Message message){
        RpcResponse received = serverExchange.received(message.getRequest());
        message.setResponse(received);
    }
    /**
     * 服务端启动成功后置处理
     * @param message
     */
    private void initServerPostProcess(Message message){
        String address = message.payload(RpcDict.SERVER_ADDRESS).toString();
        RpcDict.NodeStatus status = (RpcDict.NodeStatus)message.payload(RpcDict.SERVER_STATUS);
        //构建Rpc服务方对方并发布消息
        DefaultServer server = new DefaultServer(address,status);
        serverExchange.publishServer(server);
    }
    /**
     * 客户端连接后事件处理
     * @param message
     */
    private void reconnectPostProcess(Message message){
        InetSocketAddress address = (InetSocketAddress)message.payload(RpcDict.SERVER_ADDRESS);
        try {
            clientExchange.connection(address);
        } catch (Exception e) {
            throw new RpcRunException(e);
        }
    }
}
